import rx.Observable;
import rx.Producer;
import rx.Subscriber;
import rx.internal.producers.ProducerArbiter;
import rx.observers.TestSubscriber;

import static rx.Observable.range;

public class _Producer {
    public static final class ThenObserve<T> implements Observable.Operator<T, T> {
        final Observable<? extends T> other;

        public ThenObserve(Observable<? extends T> other) {
            this.other = other;
        }

        @Override
        public Subscriber<? super T> call(Subscriber<? super T> child) {
            ProducerArbiter arbiter = new ProducerArbiter();

            Subscriber<T> parent = new Subscriber<T>() {
                @Override
                public void onNext(T t) {
                    child.onNext(t);
                    arbiter.produced(1);
                }

                @Override
                public void onError(Throwable e) {
                    arbiter.setProducer(null);
                    child.onError(e);
                }

                @Override
                public void onCompleted() {
                    arbiter.setProducer(null);
                    //产生新的subscriber用来接收producer的改变
                    DelegatingSubscriber<T> subscriber = new DelegatingSubscriber<>(child, arbiter);
                    child.add(subscriber);
                    other.unsafeSubscribe(subscriber);
                }

                @Override
                public void setProducer(Producer p) {
                    arbiter.setProducer(p);
                }
            };
            child.add(parent);
            child.setProducer(arbiter);
            return parent;
        }
    }


    static class DelegatingSubscriber<T> extends Subscriber<T>{
        Subscriber<? super  T> actual;
        ProducerArbiter arbiter;


        public DelegatingSubscriber(Subscriber<? super  T> downstream,ProducerArbiter arbiter){
            this.actual = downstream;
            this.arbiter = arbiter;
        }
        @Override
        public void onCompleted() {
            arbiter.setProducer(null);
            this.actual.onCompleted();
        }

        @Override
        public void onError(Throwable e) {
            arbiter.setProducer(null);
            this.actual.onError(e);
        }

        @Override
        public void onNext(T o) {
            this.actual.onNext(o);
            arbiter.produced(1);
        }

        @Override
        public void setProducer(Producer p) {
           arbiter.setProducer(p);
        }
    }

    public static void main(String[] args) {
        Observable<Integer> source =range(1, 10)
                .lift(new ThenObserve<>(range(11, 90)));


        TestSubscriber<Integer> ts = new TestSubscriber<>();
        ts.requestMore(20);

        source.subscribe(ts);

        ts.getOnNextEvents().forEach(System.out::println);
    }
}

